package com.qqt.csr.ixport.service;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Lists;
import com.qqt.csr.common.acl.aics.AICustomerServiceClient;
import com.qqt.csr.common.acl.aics.resp.CsMemberInfoDTO;
import com.qqt.csr.common.exception.StatusCode;
import com.qqt.csr.common.session.SessionContextHolder;
import com.qqt.csr.common.utils.ServiceAssert;
import com.qqt.csr.ixport.assembler.BatchImportTaskAssembler;
import com.qqt.csr.ixport.entity.BatchImportTask;
import com.qqt.csr.ixport.entity.ImportTopicConfig;
import com.qqt.csr.ixport.enums.TaskStatus;
import com.qqt.csr.ixport.enums.TaskType;
import com.qqt.csr.ixport.mapper.BatchImportTaskMapper;
import com.qqt.csr.ixport.vo.req.CreatingImportTaskReqVO;
import com.qqt.csr.ixport.vo.req.QueryingImportTaskReqVO;
import com.qqt.csr.ixport.vo.req.SwitchingImportTaskStatusReqVO;
import com.qqt.csr.ixport.vo.resp.ImportTaskInfoRespVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * @Author: yang hao
 * @Date: 2025/1/20 9:24
 * @Description: 批量导入任务服务
 **/
@Slf4j
@Service
public class BatchImportTaskService extends ServiceImpl<BatchImportTaskMapper, BatchImportTask> {

    private final List<TaskStatus> processingStatusList = List.of(TaskStatus.UN_START,
            TaskStatus.RUNNING, TaskStatus.PAUSE, TaskStatus.ROLLING_BACK);
    @Autowired
    private ImportTopicConfigService importTopicConfigService;
    @Autowired
    private AICustomerServiceClient aiCustomerServiceClient;

    /**
     * 创建批量导入任务
     *
     * @param reqVO 批量导入任务新增请求参数
     * @return 批量导入任务新增响应参数
     */
    public ImportTaskInfoRespVO createTask(String tenantId, CreatingImportTaskReqVO reqVO) {
        ImportTopicConfig topicConfig = importTopicConfigService.getByBusinessCode(reqVO.getBusinessCode());
        ServiceAssert.notNull(topicConfig, StatusCode.Common.NOT_EXIST.getCode(), "配置异常！");
        ServiceAssert.isTrue(StringUtils.isNotBlank(topicConfig.getTopic()), StatusCode.Common.NOT_EXIST.getCode(), "配置异常！");

        List<BatchImportTask> existList = this.lambdaQuery().select()
                .eq(BatchImportTask::getTenantId, tenantId)
                .eq(BatchImportTask::getDataSourceDownUrl, reqVO.getDataSourceDownUrl())
                .list();
        BatchImportTask existImportTask = existList.stream().filter(task -> processingStatusList.contains(TaskStatus.match(task.getStatus())))
                .findFirst().orElse(null);
        if (existImportTask != null) {
            return BatchImportTaskAssembler.convert(existImportTask, SessionContextHolder.getHeaderInfo().getNickName());
        }

        BatchImportTask entity = BatchImportTaskAssembler.convert(reqVO);
        this.save(entity);
        return BatchImportTaskAssembler.convert(entity, SessionContextHolder.getHeaderInfo().getNickName());
    }

    /**
     * 获取任务列表
     */
    public List<ImportTaskInfoRespVO> list(String tenantId, Long csId, QueryingImportTaskReqVO reqVO) {
        String type = Optional.ofNullable(reqVO).map(QueryingImportTaskReqVO::getType).map(TaskType::getCode).orElse(null);
        List<Integer> taskStatusList = Optional.ofNullable(reqVO).map(QueryingImportTaskReqVO::getStatusList).orElseGet(Lists::newArrayList)
                .stream().map(TaskStatus::getCode).collect(Collectors.toList());
        List<BatchImportTask> tasks = this.lambdaQuery().select(BatchImportTask::getId, BatchImportTask::getName,
                        BatchImportTask::getType, BatchImportTask::getStatus, BatchImportTask::getDataSourceDownUrl,
                        BatchImportTask::getDataTotal, BatchImportTask::getSuccessCount, BatchImportTask::getFailReason)
                .eq(BatchImportTask::getTenantId, tenantId)
                .eq(BatchImportTask::getCreatorId, csId)
                .eq(StringUtils.isNotBlank(type), BatchImportTask::getType, type)
                .in(CollectionUtils.isNotEmpty(taskStatusList), BatchImportTask::getStatus, taskStatusList)
                .list();
        Map<Long, CsMemberInfoDTO> memberInfo = aiCustomerServiceClient.queryByIdList(Lists.newArrayList(csId))
                .stream().collect(Collectors.toMap(CsMemberInfoDTO::getUserId, v -> v, (o, n) -> o));
        return tasks.stream().map(task -> BatchImportTaskAssembler.convert(task, memberInfo.get(task.getCreatorId()))).toList();
    }

    /**
     * 获取任务信息
     *
     * @param tenantId
     * @param creatorId
     * @param taskId
     * @return
     */
    public ImportTaskInfoRespVO getTaskInfo(String tenantId, Long creatorId, Long taskId) {
        BatchImportTask task = this.lambdaQuery().select(BatchImportTask::getId, BatchImportTask::getName,
                        BatchImportTask::getType, BatchImportTask::getStatus, BatchImportTask::getDataSourceDownUrl,
                        BatchImportTask::getDataTotal, BatchImportTask::getSuccessCount, BatchImportTask::getFailReason)
                .eq(BatchImportTask::getId, taskId)
                .eq(BatchImportTask::getTenantId, tenantId)
                .eq(BatchImportTask::getCreatorId, creatorId).one();
        Map<Long, CsMemberInfoDTO> memberInfo = aiCustomerServiceClient.queryByIdList(Lists.newArrayList(creatorId))
                .stream().collect(Collectors.toMap(CsMemberInfoDTO::getUserId, v -> v, (o, n) -> o));
        return BatchImportTaskAssembler.convert(task, memberInfo.get(task.getCreatorId()));
    }

    /**
     * 根据状态查询任务列表
     *
     * @param lastId
     * @param batchSize 每次查询数量
     * @return 任务列表
     */
    public List<BatchImportTask> fetchUnStartStatus(Long lastId, int batchSize) {
        return this.lambdaQuery()
                .gt(BatchImportTask::getId, lastId)
                .eq(BatchImportTask::getStatus, TaskStatus.UN_START.getCode())
                .last("limit " + batchSize)
                .list();
    }

    /**
     * 更新任务进行中
     *
     * @param task 任务
     * @return 是否更新成功
     */
    public boolean updateTaskRunning(BatchImportTask task) {
        task.setStatus(TaskStatus.RUNNING.getCode());
        task.setFailReason(StringUtils.EMPTY);
        task.setRetryTimes(0);
        return this.lambdaUpdate().set(BatchImportTask::getStatus, task.getStatus())
                .set(BatchImportTask::getRetryTimes, task.getRetryTimes())
                .set(BatchImportTask::getUpdateTime, new Date())
                .set(BatchImportTask::getFailReason, task.getFailReason())
                .eq(BatchImportTask::getId, task.getId())
                .eq(BatchImportTask::getStatus, TaskStatus.UN_START.getCode())
                .update();
    }

    /**
     * 更新任务成功
     *
     * @param task 任务
     * @return 是否更新成功
     */
    public boolean updateTaskSuccess(BatchImportTask task) {
        task.setStatus(TaskStatus.SUCCESS.getCode());
        return this.lambdaUpdate().set(BatchImportTask::getStatus, task.getStatus())
                .set(BatchImportTask::getRetryTimes, task.getRetryTimes())
                .set(BatchImportTask::getDataTotal, task.getDataTotal())
                .set(BatchImportTask::getSuccessCount, task.getSuccessCount())
                .set(BatchImportTask::getUpdateTime, new Date())
                .set(StringUtils.isNotBlank(task.getFailReason()), BatchImportTask::getFailReason, task.getFailReason())
                .eq(BatchImportTask::getId, task.getId())
                .update();
    }

    /**
     * 更新任务失败
     *
     * @param task 任务
     * @return 是否更新成功
     */
    public boolean updateTaskFail(BatchImportTask task) {
        task.setStatus(TaskStatus.FAIL.getCode());
        return this.lambdaUpdate().set(BatchImportTask::getStatus, task.getStatus())
                .set(BatchImportTask::getRetryTimes, task.getRetryTimes())
                .set(BatchImportTask::getUpdateTime, new Date())
                .set(StringUtils.isNotBlank(task.getFailReason()), BatchImportTask::getFailReason, task.getFailReason())
                .eq(BatchImportTask::getId, task.getId())
                .update();
    }

    /**
     * 更新任务处理数量
     *
     * @param task
     * @return
     */
    public boolean updateTaskCount(BatchImportTask task) {
        return this.lambdaUpdate()
                .set(BatchImportTask::getDataTotal, task.getDataTotal())
                .set(BatchImportTask::getSuccessCount, task.getSuccessCount())
                .set(BatchImportTask::getUpdateTime, new Date())
                .eq(BatchImportTask::getId, task.getId())
                .update();
    }

    /**
     * 是否停止任务
     *
     * @param taskId
     * @return
     */
    public boolean isInterruptTask(Long taskId) {
        return this.lambdaQuery().eq(BatchImportTask::getId, taskId)
                .in(BatchImportTask::getStatus, TaskStatus.SUCCESS.getCode(), TaskStatus.FAIL.getCode(),
                        TaskStatus.CANCEL.getCode(), TaskStatus.PAUSE.getCode(), TaskStatus.ROLLING_BACK.getCode(),
                        TaskStatus.ROLLBACK_FAIL.getCode())
                .count() > 0;
    }

    /**
     * 切换任务状态
     *
     * @param tenantId
     * @param reqVO
     */
    public void switchTaskStatus(String tenantId, Long creatorId, SwitchingImportTaskStatusReqVO reqVO) {
        Long taskId = reqVO.getTaskId();
        TaskStatus nStatus = reqVO.getNewStatus();
        List<TaskStatus> oStatusList = getSwitchOldStatus(nStatus);
        ServiceAssert.isTrue(CollectionUtils.isNotEmpty(oStatusList), StatusCode.Common.CUSTOMIZE_ERROR_MESSAGE.getCode(), "无效的状态切换！");
        boolean result = this.lambdaUpdate().set(BatchImportTask::getStatus, nStatus.getCode())
                .eq(BatchImportTask::getId, taskId)
                .eq(BatchImportTask::getTenantId, tenantId)
                .eq(BatchImportTask::getCreatorId, creatorId)
                .in(BatchImportTask::getStatus, oStatusList.stream().map(TaskStatus::getCode).collect(Collectors.toSet()))
                .update();
        ServiceAssert.isTrue(result, StatusCode.Common.CUSTOMIZE_ERROR_MESSAGE.getCode(), "无效的状态切换！");
    }

    private List<TaskStatus> getSwitchOldStatus(TaskStatus newStatus) {
        return switch (newStatus) {
            case UN_START -> Lists.newArrayList(TaskStatus.PAUSE);
            case PAUSE -> Lists.newArrayList(TaskStatus.UN_START, TaskStatus.RUNNING);
            case CANCEL -> Lists.newArrayList(TaskStatus.UN_START, TaskStatus.RUNNING, TaskStatus.PAUSE);
            default -> Lists.newArrayList();
        };
    }

}
